package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/3/4
 *
 *  upsert-kafka：  把一个会发生更新编变化的表中的数据写出到kafka。
 *          用普通的格式，写 insert(+I) 和 update-after(+U)后的数据。
 *          用value=null,key=主键这种格式，表示删除的数据。
 *          用value=新值，key=主键这种格式，标识更新的数据。
 *
 *          举例:  id=s1,vc=20
 *                      第一次写入:  { key=s1,value = { id=s1,vc=20 }  }
 *                      更新后写入:   { key=s1,value = { id=s1,vc=30 }  }
 *                      删除:        { key=s1,value = {}  }
 *
 *
 *                      不能指定sink的方式，因为数据有key，按照key进行hash分区，相同key的数据分在同一个分区.
 *                      定义表时，要有主键约束
 *
 *
 *  和hbase原理一样。  hbase依赖HDFS进行存储，HDFS不支持随机写，只支持追加写。
 *          写更新和删除时，是 append一条记录，这条记录的类型是 update或delete类型。
 *
 *
 *
 *
 */
public class Demo6_WriteKafkaAgg
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());

        //为流起名字
        tableEnvironment.createTemporaryView("source",ds);

        //使用create语句创建表，映射文件系统中指定的数据
        String createTableSql = " create table t1 ( id string   PRIMARY KEY ,sumVc double  )" +
                                " with ( " +
                                 " 'connector' = 'upsert-kafka' ,   " +
                                 " 'topic' =  'topicC'  ," +
                                 " 'properties.bootstrap.servers' = 'hadoop102:9092'," +
                                 "  'key.format' = 'json' ," +
                                 "  'value.format' = 'json' " +
                                 "      )                 ";

        //执行建表
        tableEnvironment.executeSql(createTableSql);

        // 查询语句是一个聚合操作，表中的数据会发生更新， 普通的kafka连接器，无法处理这种更新。 需要选择upsert-kafka连接器
        tableEnvironment.executeSql(" insert into  t1 select id,sum(vc) from source group by id ");



    }
}
